<?php
namespace Handler;

use Core\Model;
use Core\TbRedis;

class WebSocket
{
    private $client;
    private $redis;
    private $db;
    private $swoole_serialize;
    protected $ouibase;

    protected $taskProcess;
    protected $cpu_num;
    private $processTable;
    private $Counter_timer;
    /**
     * 构造函数
     */
    public function __construct()
    {
        $this->cpu_num = WS_CPU_NUM; //swoole_cpu_num();
        $this->client = new \Swoole\table(1024);
        $this->client->column('fd', \Swoole\table::TYPE_INT, 8);
        $this->client->column('type', \Swoole\table::TYPE_STRING, 64);
        $this->client->column('deviceID', \Swoole\table::TYPE_STRING, 128);//探霸
        $this->client->column('tbid', \Swoole\table::TYPE_INT, 64);
        $this->client->column('datatype', \Swoole\table::TYPE_STRING, 16);
        $this->client->column('devmac', \Swoole\table::TYPE_STRING, 64);   //订阅频道
        $this->client->column('channel', \Swoole\table::TYPE_STRING, 64);   //订阅频道
        $this->client->column('stop', \Swoole\table::TYPE_INT, 2);    //关闭状态
        $this->client->column('encode', \Swoole\table::TYPE_INT, 2);    //加密状态
        $this->client->column('interval', \Swoole\table::TYPE_INT, 8);
        $this->client->column('multi', \Swoole\Table::TYPE_STRING, 4);
        $this->client->create();

        $this->swoole_serialize = new \Swoole\Serialize();

        $this->processTable = new \Swoole\table(128);/*进程pid表，键为进程pid*/
        $this->processTable->column('stt', \Swoole\table::TYPE_INT,2);
        $this->processTable->create();

        //oui内存对照库- key是Assignment  value --uid
        $this->ouibase = new  \Swoole\table(65535);
        $this->ouibase->column('uid', \Swoole\table::TYPE_INT, 8);
        $this->ouibase->column('ouiname', \Swoole\table::TYPE_STRING, 200);
        $this->ouibase->create();

        $this->redis=TbRedis::init("wbscktinit0");
        $this->db =  Model::init("wbscktinit0");

    }

    /**
     * 服务器关闭函数
     * @param $serv
     */
    public function onShutdown( $serv )
    {
        \Swoole\Process::kill($serv->manager_pid,0);

        for ($i = 0; $i < $this->cpu_num ; $i++) {
            //  foreach($this->processTable as $k){
            \Swoole\Process::kill($this->taskProcess[$i]);

            $this->processTable->del($this->taskProcess[$i]);
        }
    }

    public function onTask($serv, $task_id, $from_id, $data )
    {   }

    public function onFinish($serv, $task_id, $data)
    {   }

    /**
     * 主进程启动回调方法
     * @param  Swoole\Server $serv    swoole对象
     */
    public function onStart($serv)
    {
        cli_set_process_title("php swoole: websocketserver master Server");
        echo "php swoole: websocket Server start...";

        $datas = $this->db->select('oui',  ["id", "Assignment","OrganizationName"]);
        if(!empty($datas)){
            foreach($datas as $item)
            {
                $this->ouibase->set(strtoupper($item['Assignment']), array('uid' => $item['id'],'ouiname'=>$item['OrganizationName']));
            }
        }

    }


    public function onManagerStart($serv)
    {

        $serv->cmdProc = new \Swoole\Process(function ($process) use($serv) {
            cli_set_process_title('websocketserver main Process');

            $serverconfig = config('tcpserver');
            $ip = $serverconfig['host'];
            $port = $serverconfig['port'];
            $serv->cli = new \Swoole\Client(SWOOLE_SOCK_TCP);
            $serv->cli->set(array(
                'open_eof_check' => true,
                'package_eof' => PHP_EOL,
            ));
            while (1) {
                try {
                    while (1) {
                        try {
                            if (false == @$serv->cli->connect($ip, $port, -1)) {
                                echo "连接{$port}服务器失败\n";
                                throw new \Exception("connect faild");
                            }
                            $serv->cli->send('{"command": "getdata","cname": "","from":"websocket"}' . PHP_EOL);
                            echo "连接{$port}服务器成功\n";
                            break;
                        } catch (\Exception $e) {
                            $serv->cli->close();
                            echo "正在尝试重新连接服务器\n";
                            sleep(5);
                        }
                    }

                    while (1) {
                        try {
                            if (false ==  $Redis = @TbRedis::init('webscktcp')) {
                                echo "Redis connect faild! \n";
                                throw new \Exception("Redis connect faild");
                            }

                            break;
                        } catch (\Exception $e) {
                            echo "reconnect Redis …… \n";
                            sleep(2);
                        }
                    }

                    while ($data = $serv->cli->recv()) {
                        $data = json_decode($data, true);
//                        var_dump($data);
                        if (!is_array($data)) continue;
                        $data['time'] = date('Y-m-d H:i:s', time());
                         echo ">";

                        if (in_array($data['type'], ['a000', 'a001', 'a002','b000'])) {
                            //随机丢到进程处理
                            $taskId = mt_rand(0, ($this->cpu_num - 1));
                            $this->taskProcess[$taskId]->push($this->swoole_serialize::pack($data));

                        }


                    }
                    //                   });
                } catch (\Exception $e) {
                    echo "连接出现错误\n";
                    $serv->cli->close();
                }
            }
        });
        $serv->cmdProc->start();


        //定时，在没有数据的情况下发送计数结果
        //每个链接遍历
        /*******************************************进程不定时清理redis******************************************/
        (new \Swoole\Process(function ($process) use($serv) {
            cli_set_process_title('basedataserver_counttimer_Process');

            while (1) {
                try {
                    if (false ==  $redis = @TbRedis::init('counttimerProcess')) {
                        echo "Redis connect faild! \n";
                        throw new \Exception("Redis connect faild");
                    }

                    break;
                } catch (\Exception $e) {
                    echo "reconnect Redis …… \n";
                    sleep(2);
                }
            }
            while (1) {
                sleep(5);
                foreach ($this->client as $row) {
                    if ($serv->exist($row['fd']) && isset($row['deviceID'])) {

                        $info = $redis->hGetAll($row['deviceID']);

                        $ncount0 = isset($info['a000.count5']) ? $info['a000.count5'] : 1;
                        $ncount1 = isset($info['a001.count5']) ? $info['a001.count5'] : 1;
                        $ncount2 = isset($info['a002.count5']) ? $info['a002.count5'] : 1;
                        $ncount= $ncount0+$ncount1+$ncount2;
                        if ($row['type'] == 'getdata' && !empty($row['deviceID'])) {
                            if (!isset($row['devmac']) || empty($row['devmac'])) {
                                $serv->push($row['fd'],json_encode(array(
                                    'type' => 'devcount',
                                    'data' => $ncount,
                                    'dataa000' => $ncount0,
                                    'dataa001' => $ncount1,
                                    'dataa002' => $ncount2,
                                    'from'=>'timer'
                                )));

                            }
                        }

                    }

                }
            }


        }, false, false))->start();


        //读取RF设置紧急状态redis命令进程
        ( new \Swoole\Process(function ($process) use($serv) {
            while (true) {
                while (true) {
                    try {
                        $redis =TbRedis::init("websktugtnotice");
                        break;
                    } catch (\RedisException $e) {
                        echoMsg('redis websocket urgentstatus notice connect fail', $e->getCode(), $e->getLine());
                        sleep(2);
                    }
                }
// httpserver  $ret= $this->redis->lPush('rfurgent', json_encode(array( 'tbmac'=>$param['tbmac'],'fd'=>$fd,'tagID'=>$param['devmac'],'time'=>time())));
                while (true) {
                    try {
                    	// 从redis中读取紧急RF队列数据
                        $data = json_decode($redis->brPop('rfurgent', 0)[1], true);
                        if(is_array($data)){
                            foreach ($this->client as $row) {
                                if ($serv->exist($row['fd']) && $row['type'] == 'getdata' ) {
                                    $serv->push($row['fd'],json_encode(array(
                                        'type' => 'rfurgent',
                                        'tbmac'=>$data['tbmac'],
                                        'fd'=>$data['fd'],
                                        'devmac'=>$data['tagID'],
                                        'dotime'=>date('Y-m-d H:i:s',$data['time'])
                                    )));

                                    $redis->del("UrRF_".$data['tagID']);
                                }

                            }
                        }

                    } catch (\RedisException $e) {
                        echoMsg('redis  dataserver commander error [' . $e->getMessage() . ']', $e->getCode(), $e->getLine(),$redis);
                        break;
                    }

                }
            }
        }, false))->start();

        //设备状态推送进程
        ( new \Swoole\Process(function ($process) use($serv) {
            while (true) {
                //连接redis
                while (true) {
                    try {

                        $theRedis =TbRedis::init("tbcommderstatus");
                        break;
                    } catch (\RedisException $e) {
                        Log::print('redis connect fail', $e->getCode(), $e->getLine());
                        sleep(2);
                    }
                }

                while (true) {
                    $theRedis->psubscribe([getDbNum() . REDIS_TB_STATUS_ALL], function ($redis, $chan, $chanName, $data) use ($serv) {
                        $data = json_decode($data, true);

                                $sdata=array('deviceID' => $data['deviceid'],'data'=>$data);
                                $tbserver='config';
                                if($chanName== getDbNum() . REDIS_TB_STATUS_CONFIG){
                                    $tbserver='config';//命令服务器
                                    $sdata['cfgstatus']  = $data['status'];
                                }
                                elseif($chanName== getDbNum() . REDIS_TB_STATUS_SERVER){
                                    $tbserver='data';//数据服务器
                                    $sdata['datastatus']  = $data['status'];

//                                    $data['time'] = date('Y-m-d H:i:s', $data['time']);

                                    $online = null;
                                    $online[] = array(
                                        'deviceID' => $data['deviceid'],
                                        'status' =>(isset($data['status']) && $data['status'])==1?1:0
                                    );
                                    //设备下线信息
                                    $pushData = json_encode(array('type' => 'Status', 'data' => $online));

                                    foreach ($this->client as $row) {
                                        if(isset($row['fd'])&& !empty($row['fd']) && $serv->exist($row['fd'])){
                                            if (isset($row['deviceID']) && isset($data['deviceid']) && $row['deviceID'] == $data['deviceid'])
                                                $serv->push($row['fd'], $pushData);
                                            else
                                                $serv->push($row['fd'], $pushData);
                                        }

                                    }

                                }
                    });
                }
            }
        }, false,false))->start();


    }

    /**
     * worker进程启动回调方法
     * @param  Swoole\Server $serv    swoole对象
     * @param  int $worker_id         worker进程id号
     */
    public function onWorkerStart($serv, $worker_id)
    {

    }

    /**
     * 连接回调方法
     * @param  Swoole\Server $serv    swoole对象
     * @param  int $req               连接句柄编号
     */
    public function onOpen($serv, $req)
    {
        //  var_dump( $serv->getClientInfo($req->fd));//$serv->exist($req->fd),
        $this->client->set($req->fd, array('encode' => 0));
        if($req->get)
        {
            if(isset($req->get['dcdlzd']))
                $this->client->set($req->fd, array('encode' => 1));
        }

        $this->setMsgClients($serv);
    }

    /**
     * 接收数据回调方法
     * @param  Swoole\Server $serv    swoole对象
     * @param  int $frame             连接句柄编号
     */
    public function onMessage($serv, $frame)
    {
        $data = json_decode($frame->data, true);
        $redis =TbRedis::init("tbwebsctsv");

        //消息类型
        if (!isset($data['type'])) {
            $serv->push($frame->fd, 'type is null');
            return;
        }

        if($data['type'] == 'Status') {

            $result = $redis->keys('DZ*');
            $list = array();
            foreach ($result as $r) {

                if (! strpos($r, ".") ) {
                    $info = $redis->hGetAll($r);
                    if (isset($info['fd']) && $info['fd'] > 0) $list[] = array('deviceID' => $info['mac'], 'status' => $info['status']);
                }
            }

            $serv->push($frame->fd, json_encode(array('type'=> 'Status', 'data' => $list)));

        }

        $this->client->set($frame->fd, array('fd' => $frame->fd, 'type' => $data['type'],'deviceID' => '','devmac' =>'','channel'=>'','stop' => '','datatype'=>'','interval'=>0));

        if(isset($data['device']) && !empty($data['device']) ){

            $tbid=$redis->hGet($data['device'],'id');
            if(!isset($tbid) || empty($tbid))
                $tbid=$this->db->get('device','id',['mac'=>$data['device']]);

            if(!isset($tbid) || empty($tbid))
            {
                $serv->push($frame->fd, json_encode(array('type'=> 'erro', 'data' => [],'msg'=>'tb isnotfond')));
                return;
            }


            if($data['type'] == 'online') {
                $info = $redis->hGet($data['device'],'status');

                $list =  array('deviceID' => $data['device'], 'status' => $info);
                $serv->push($frame->fd, json_encode(array('type'=> 'online', 'time'=>time(),'data' => $list)));
                return;

            }
            else if ($data['type'] == 'getdata'){
                //专用订阅频道
                $type=!isset($data['datatype'])?"STA": DzDataType($data['datatype']);
                $datatype=!isset($data['datatype'])?"all": $data['datatype'];
                $data['cname']="base_data_{$type}_{$tbid}" ;
                $this->client->set($frame->fd, array('channel' => $data['cname'],'stop' => 'n','datatype'=>$datatype));
                echo " fd[{$frame->fd}] start TCP :" . $data['cname'] ." ".date('Y-m-d H:i:s', time())."\n" ;

                if(isset($data['devmac']))
                    $this->client->set($frame->fd, array('devmac' => $data['devmac']));
            }
            else if ($data['type'] == 'dataEnd'){
                //专用订阅频道
                $type=!isset($data['datatype'])?"STA": DzDataType($data['datatype']);
                $datatype=!isset($data['datatype'])?"all": $data['datatype'];
                $data['cname']="base_data_{$type}_{$tbid}";
                $this->client->set($frame->fd, array('channel' => $data['cname'],'stop' => 'y','datatype'=>$datatype));
                echo " fd[{$frame->fd}] close TCP :" . $data['cname'] ." ".date('Y-m-d H:i:s', time())."\n" ;

            }else if ($data['type'] == 'history'){
                $ifencode= $this->client->get($frame->fd, 'encode');
                $datatype=!isset($data['datatype'])?"a001": $data['datatype'];
//                echo " fd[{$frame->fd}]:{$data['device']} get history {$datatype}" . date('Y-m-d H:i:s', time())."\n" ;

                $total=0;
                $maclist=[];

                //最近5分钟之内的统计--都健在的
                foreach ($redis->keys($datatype.'.*.'.$tbid.".DevTbExpired".PRERDS) as $ky){
                    $rdkey=explode(".",$ky);
                    $macl['devmac']=$rdkey[1];

                    if($datatype =="a000" )
                        $devid= $redis->hget('AP_'.$rdkey[1], 'devid');
                    elseif($datatype =="a002" )
                        $devid=  $redis->hget('RF_'.$rdkey[1], 'devid');
                    elseif($datatype =="a001" )
                        $devid=$redis->hget('STA_'.$rdkey[1], 'devid');
                    else
                        continue;

                    $macl['deviceid']=$devid;

                    $rdsmKey=explode(".DevTbExpired".PRERDS,$ky);
                    $tmdata=$redis->hgetall($rdsmKey[0]);
                    $macl['createtime']=$tmdata['time'];//时间出现bug，非正常时间戳-注意排查
                    $macl['data']=$tmdata['data'];
                    $macl['rssi']=$tmdata['rssi'];
                    $maclist[]=$macl;
                    $total++;
                }

                $serv->push($frame->fd,json_encode(array('type'=> 'historylist','total'=>$total))) ;

                if(!empty($maclist)){
                    foreach ($maclist as $mr){
                        if(!isset($mr['deviceid']) || !isset($mr['devmac']) || !isset($mr['createtime'])) continue;
                        $tdevmac=$mr['devmac'];
                        $tbdata=json_decode($mr['data'],true);

                        $ifurgrntrf=0;
                        if($datatype=='a002' && $redis->exists("UrRF_".$tdevmac)){
                            $ifurgrntrf= $redis->hGet("UrRF_".$tdevmac,'urgentStatus');
                        }
                        $ifurgrntrf=isset($ifurgrntrf)&& empty($ifurgrntrf) ?$ifurgrntrf:0;

                        $row = array(
                            'tbid'     => $tbid,
                            'tbmac'    =>  $data['device'],
                            'type'     => $datatype,
                            'devkey'   => 1==$ifencode?$tdevmac:md5("dztb".$tdevmac),
                            'dv_brand' => $datatype =="a001"?$tbdata['dv_brand']:"未知",
                            'ap_brand' => $datatype =="a001"?$tbdata['ap_brand']:"未知",
                            'time'     =>  date('Y-m-d H:i:s',$tbdata['time']),//$tbdata['time']<100? date("Y-m-d H:i:s"):
                            'mac'      => $data['device'],
                            'rssi'  => isset($data['rssi'])? -$data['rssi']:-100,
                            'sendRssi'=>isset( $data['sendRssi'])?-$data['sendRssi']:-100,
                            'find'     =>$redis->hGet($datatype.'.'.$tdevmac.".".$tbid,'count'),
                            'urgentStatus'=>$datatype=='a002'?$ifurgrntrf:0 //历史的紧急状态获取有问题，

                        );
                        $row['sendRssi']=isset( $data['sendRssi'])? (-$data['sendRssi']):$row['rssi'];
                        if(isset( $tdevmac)&& !empty($tdevmac))
                            $serv->push($frame->fd,json_encode(array('type'=> 'history', 'data' => $row))) ;

                    }
                }

                $serv->push($frame->fd,json_encode(array('type'=> 'historyend'))) ;
                return;
            }
            $this->client->set($frame->fd, array('fd' => $frame->fd, 'type' => $data['type'],'deviceID' =>  $data['device']));
        }


        $this->setMsgClients($serv);
    }

    /**
     * 关闭回调方法
     * @param  Swoole\Server $serv    swoole对象
     * @param  int $fd                连接句柄编号
     */
    public function onClose($serv, $fd, $reactorId)
    {

        $this->client->del($fd);
        $this->setMsgClients($serv);
    }

    public function onWorkerStop($serv, $worker_id)
    {}

    public function onManagerStop($serv)
    { }

    /**
     * 格式化mac地址
     * @param $mac    mac地址
     * @return string 格式化后的mac地址
     */
    protected function formatMac($mac)
    {
        return implode(':', str_split(mb_strtoupper($mac), 2));
    }


    public function getProcess($serv)
    {
        for ($i = 0; $i < $this->cpu_num ; $i++) {
            $this->taskProcess[$i] = new \Swoole\Process(function ($process) use ($serv,$i){
                $redis=TbRedis::init("wbscktask.".$i);
//                $thisdb=Model::init("wbsckpro.".$i);
                echo "websocket server data process[{$i}] start . \n";

                while (true) {
                    $data = $process->pop();

                    $data=$this->swoole_serialize::unpack($data);
                    if(empty($data)) continue ;

                    if (isset($data['type']) && in_array($data['type'], ['a000', 'a001', 'a002','b000'])) {

                        foreach ($this->client as $row) {
                            if($serv->exist($row['fd'])) {
                                if ($row['type'] == 'getdata' && $row['deviceID']==$data['tbmac']) {

                                    if(isset($row['devmac']) && !empty($row['devmac'])){
                                        $row['datatype']=!isset($row['datatype'])?'a001':$row['datatype'];//发送指定设备信息

                                        if($data['devkey'] !==$row['devmac']) continue;

                                    }

                                    $row['datatype']=isset($row['datatype'])?$row['datatype']:"all";
                                    if($row['stop']=='n' && ($row['datatype']=="all" || $data['type']==$row['datatype'])) {

                                            $devrkey=$data['type'].".".$data['devkey'] . "." . $data['tbid'] ;
                                            $finds=$redis->hGet($devrkey,'count');

                                            $ifurgrntrf=0;
                                            if($data['type']=='a002' && $redis->exists("UrRF_".$data['devkey'])){
                                                $ifurgrntrf=$redis->hGet("UrRF_".$data['devkey'],'urgentStatus');
                                            }

                                            $pushdata=array(
                                                'tbid'=>$data['tbid'],
                                                'tbmac'=>$data['tbmac'],
                                                'type'=>$data['type'],
                                                'devkey'=> 1==$row['encode'] ?$data['devkey']:md5("dztb".$data['devkey']),//strrev($data['devkey'])
                                                'rssi'=>$data['rssi'],
                                                'sendRssi'=>isset( $data['sendRssi'])?$data['sendRssi']:$data['rssi'],
                                                'find'=>isset($finds)?$finds:1,
                                                'devid'=>$data['devid'],
                                                'ap_brand' =>isset( $data['ap_brand'])?$data['ap_brand']:'未知',
                                                'dv_brand' =>isset( $data['dv_brand'])?$data['dv_brand']:'未知',
                                                'mac'      => 1==$row['encode'] ?$data['devkey']:md5("dztb".$data['devkey']), //strrev($data['devkey'])
                                                'time'=>$data['time'],
                                                'urgentStatus'=>$data['type']=='a002'? $ifurgrntrf:0 ,
                                                'sfind'=>isset($sfinds)?$sfinds:1,
                                            );

                                        $serv->push($row['fd'], json_encode(array('type' => 'Data', 'data' => $pushdata)));//广播

                                    }

                                }
                                elseif ($row['type'] == 'getdata' && (!isset($row['deviceID']) || empty($row['deviceID'])) ) {
                                    $serv->push($row['fd'], json_encode(array('type' => 'Data', 'data' => $data)));//广播
                                }
                            }

                        }
                    }

                    $ns=date('s',strtotime($data['time']));
                    $isserch=false;
                }
            });
//            $this->processTable->set($this->taskProcess[$i],array("stt"=>1));
            $this->taskProcess[$i]->useQueue();
        }

        return $this->taskProcess;
    }

    public function setMsgClients($serv)
    {
        $count1=0;
        $count2=0;
        foreach ($this->client as $row) {
            if($serv->exist($row['fd'])) {
                if ($row['type'] == 'getdata' && isset($row['deviceID']))
                    $count1++;
                elseif ($row['type'] == 'getdata' && (!isset($row['deviceID']) || empty($row['deviceID'])) ) {
                    $count2++;
                }
            }

        }

        foreach ($this->client as $row) {
            if($serv->exist($row['fd'])) {
                if ($row['type'] == 'getdata' && isset($row['deviceID'])) {
                    $serv->push($row['fd'], json_encode(array('type'=> 'clients', 'data' => ['counts1'=>$count1])));
                }
                elseif ($row['type'] == 'getdata' && (!isset($row['deviceID']) || empty($row['deviceID'])) ) {
                    $serv->push($row['fd'], json_encode(array('type'=> 'clients', 'data' => ['counts2'=>$count2])));
                }
                $serv->push($row['fd'], json_encode(array('type'=> 'clients', 'data' => ['counts'=>count($serv->connections)])));
            }
        }
//        echo "Now：".$count1."|".$count1." \n";
    }
}
